package net.bwie.realtime.jtp.dwd.logs.car.job;
import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
import net.bwie.realtime.jtp.utils.DorisUtil;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.ObjectMapper;
import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.node.ObjectNode;

import java.util.Properties;
public class One {
    public static void main(String[] args) throws Exception {
        // 1. 创建执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        env.enableCheckpointing(3000L);
        // 2. Kafka配置
        Properties kafkaProps = new Properties();
        kafkaProps.setProperty("bootstrap.servers", "node101:9092");
        kafkaProps.setProperty("group.id", "vehicle-status-group");

        // 3. 创建Kafka数据源
        DataStream<String> kafkaStream = env
                .addSource(new FlinkKafkaConsumer<>(
                        "dwd-car-logs",
                        new SimpleStringSchema(),
                        kafkaProps
                ));

        // 4. 数据转换处理
        DataStream<String> resultStream = kafkaStream.map(new RichMapFunction<String, String>() {
            private transient ObjectMapper mapper;

            @Override
            public void open(org.apache.flink.configuration.Configuration parameters) throws Exception {
                mapper = new ObjectMapper();
            }

            @Override
            public String map(String value) throws Exception {
                ObjectNode node = mapper.readValue(value, ObjectNode.class);

                String vin = node.get("vin").asText();
                int type = node.get("type").asInt();
                String chargingStatus = node.get("chargingStatus").asText();
                double speed = node.get("speed").asDouble();
                double batteryLevel = node.get("batteryLevel").asDouble();
                double longitude = node.get("longitude").asDouble();
                double latitude = node.get("latitude").asDouble();

                // 状态判断逻辑
                String status;
                if (type == 0 && "charging".equals(chargingStatus)) {
                    status = "充电中";
                } else if (type == 0) {
                    status = "启动中";
                } else {
                    status = "熄火中";
                }

                String ignitionStatus = type == 0 ? "已启动" : "已熄火";
                String chargingStatusDesc = "charging".equals(chargingStatus) ? "充电中" : "未充电";

                // 构造结果字符串
                return vin+","+status+","+ignitionStatus+","+chargingStatusDesc+","+speed+","+batteryLevel+","+longitude+","+latitude;
            }
        });

        // 7. 写入Doris
        DorisUtil.saveToDoris(
                resultStream,
                "ner_car_report",
                "Vehicle_a"
        );

        // 5. 打印结果到控制台
        resultStream.print();

        // 6. 执行任务
        env.execute("Vehicle Status Processing (No Doris)");
    }
}
